package com.atguigu.sparkcore.rdd.funtionser

import com.atguigu.sparkcore.rdd.funtionser.SerDemo.Searcher
import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 传递函数实例
  * author 剧情再美终是戏
  * mail 13286520398@163.com
  * date 2020/1/8 6:28 
  * version 1.0
  **/
object SerDemo {

  // TODO 1、传递一个类中的方法或函数时，需要对类序列化
  // TODO 2、匿名函数自身就有序列化的功能
  // TODO 3、传递值，不用要序列化
  // TODO 4、使用kryo序列化，也要继承Serialzable接口
  def main(args: Array[String]): Unit = {

    // 获取sparkContext
    val sc = MySparkContextUtil.get(args)

    // 注册序列化类
    //    conf.registerKryoClasses(Array(classOf[Searcher]))

    // 创建rdd
    val list = Array("hello", "hello2", "12", "4543", "3hello")
    val rdd = sc.makeRDD(list, 5)


    // 转换rdd
    val searcher: Searcher = new Searcher("hello")
    val result = searcher.getMaths2(rdd)


    // 输出
    val r1 = result.foreachPartition({ it =>
      it.foreach({ x =>
        println("foreachPartition " + Thread.currentThread().getName + "----------------" + x)
      })
    })

    Thread.sleep(10000000000000L)
    // 关闭资源
    MySparkContextUtil.close(sc)
  }

  class Searcher(val query: String) extends Serializable {
    //  class Searcher(val query: String) {
    // 判断等是否相等
    def isMath(s: String) = {
      println("isMath " + Thread.currentThread().getName + "----------------" + s)
      s.contains(query)
    }

    //  过滤rdd,返回新的rdd
    def getMaths(rdds: RDD[String])   = {
      rdds.filter(isMath)
    }

    // 过滤rdd,返回新的rdd,匿名函数
    def getMaths2(rdds: RDD[String]) = {
      val q = query
      rdds.filter({
        //        _.contains(q)
        println("getMaths2 " + Thread.currentThread().getName + "----------------")
        isMath
      })
    }
  }

}
